package top.whysu.edevice.global;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import top.whysu.edevice.constant.CodeConstant;
import top.whysu.edevice.exception.BusinessException;
import top.whysu.edevice.service.system.SystemConfigService;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池创建类
 */
@Service
public class ThreadPoolBuilder {

    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolBuilder.class);
    private static Map<String, ExecutorService> threadPoolMap;

    static {
        threadPoolMap = new ConcurrentHashMap<>();
    }

    @Resource
    private SystemConfigService systemConfigService;

    public static Map<String, ExecutorService> getThreadPoolMap() {
        return threadPoolMap;
    }

    public synchronized ExecutorService getThreadPool(String key) {
        if (StringUtils.isBlank(key)) {
            LOGGER.error("线程池创建传入参数为空！");
            throw new BusinessException(CodeConstant.Http.PARAM_ERROR, "线程池创建传入参数为空！");
        }
        ExecutorService executorService = threadPoolMap.get(key);
        if (executorService == null || executorService.isTerminated()) {
            String params = systemConfigService.getThreadPoolConfig(key);
            if (StringUtils.isBlank(params)) {
                LOGGER.error(key + ":线程池配置参数为空异常！");
                throw new BusinessException(CodeConstant.Http.PARAM_ERROR, key + ":线程池配置参数为空异常！");
            }
            ThreadPoolParams threadPoolParams = JSONObject.parseObject(params, ThreadPoolParams.class);
            executorService = new ThreadPoolExecutor(threadPoolParams.corePoolSize, threadPoolParams.maximumPoolSize,
                    threadPoolParams.keepAliveTime, TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(threadPoolParams.queueSize),
                    new NameTreadFactory(key),
                    new MyIgnorePolicy());
            threadPoolMap.put(key, executorService);
            LOGGER.info("{}线程池已经创建成功,当前线程池总概况{}", key, threadPoolMap);
        }
        LOGGER.debug("{}获取到线程池,当前线程池总概况{}", key, threadPoolMap);
        return executorService;
    }

    /**
     * 线程池管理
     */
    public static class MyIgnorePolicy extends ThreadPoolExecutor.DiscardOldestPolicy {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            super.rejectedExecution(r, e);
            doLog(r, e);
        }

        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志记录等
            LOGGER.info("线程池拒绝请求：Runnable info {},ThreadPoolExecutor info {}", e, r);
        }
    }

    /**
     * 线程命名工厂
     */
    static final class NameTreadFactory implements ThreadFactory {
        private final AtomicInteger mThreadNum = new AtomicInteger(1);
        /**
         * 线程类型名称
         */
        private String threadName;

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, threadName + mThreadNum.getAndIncrement());
        }

        private NameTreadFactory(String threadName) {
            this.threadName = threadName;
        }
    }

    /**
     * 线程池参数
     */
    private static class ThreadPoolParams {
        /**
         * 核心线程数
         */
        public int corePoolSize;
        /**
         * 最大线程数
         */
        public int maximumPoolSize;
        /**
         * 线程存活时间
         */
        public long keepAliveTime;
        /**
         * 队列大小
         */
        public int queueSize;
    }

    /**
     * 发送消息到设备线程池
     */
    public static final String THREADPOOL_LOCAL_KEY_PUBLISHMSG_TO_DEVICE = "THREADPOOL_LOCAL_KEY_PUBLISHMSG_TO_DEVICE";
    /**
     * 设备MQTT消息到达处理线程池
     */
    public static final String THREADPOOL_LOCAL_KEY_DEVICE_MSG_ARRIVED = "THREADPOOL_LOCAL_KEY_DEVICE_MSG_ARRIVED";
    /**
     * 设备上下线入口总线程池
     */
    public static final String THREADPOOL_DEVICE_KEY_ENTRANCE_ON_LINE = "THREADPOOL_DEVICE_KEY_ENTRANCE_ON_LINE";
    /**
     * 设备请求入口总线程池
     */
    public static final String THREADPOOL_DEVICE_KEY_ENTRANCE_REQUEST = "THREADPOOL_DEVICE_KEY_ENTRANCE_REQUEST";
    /**
     * 设备应答入口总线程池
     */
    public static final String THREADPOOL_DEVICE_KEY_ENTRANCE_RESPONSE = "THREADPOOL_DEVICE_KEY_ENTRANCE_RESPONSE";

    /**
     * APP登录
     */
    public static final String THREADPOOL_APP_LOGIN = "THREADPOOL_APP_LOGIN";
    /**
     * 报警信息请求
     */
    public static final String THREADPOOL_ALARM_INFO = "THREADPOOL_ALARM_INFO";
    /**
     * 添加摄像头/船舶请求
     */
    public static final String THREADPOOL_ADD_DEVICE = "THREADPOOL_ADD_DEVICE";
}
